In Prefect v3, normally I'd run ```from prefect.deployments import run_deployment  run_deployment(     name="custom-movements-flow/custom-movements",     parameters={"position_inches": 35.5} )```, but this is blocking. I want to be able to send a command asynchronously and poll its status for later (similar to baking an oven, preheat to 350 degrees, not sitting in front of the oven the entire time)

I'll help you find the async solution for Prefect v3. Let me fetch the documentation on deployments and async execution.


resolve-library-id
libraryName:
Prefect
Now let me fetch the Prefect documentation focused on async execution and non-blocking deployment runs:


get-library-docs
context7CompatibleLibraryID:
/prefecthq/prefect
topic:
async deployment run_deployment non-blocking
Perfect! I found exactly what you need. Here's how to run a deployment asynchronously and poll its status later in Prefect v3:

Async Non-Blocking Deployment Trigger
Instead of blocking with run_deployment(), you can use await run_deployment() in an async context:

import asyncio
from prefect.deployments import run_deployment

async def trigger_deployment():
    flow_run = await run_deployment(
        name="custom-movements-flow/custom-movements",
        parameters={"position_inches": 35.5}
    )
    return flow_run

# Trigger it without waiting
flow_run = asyncio.run(trigger_deployment())
print(f"Flow run ID: {flow_run.id}")
Return Immediately Without Waiting
If you want to trigger and return immediately (set the oven to 350° and walk away), use timeout=0:

from prefect.deployments import run_deployment

flow_run = run_deployment(
    name="custom-movements-flow/custom-movements",
    parameters={"position_inches": 35.5},
    timeout=0  # Returns immediately
)
print(f"Flow run ID: {flow_run.id}")
Polling the Status Later
Once you have the flow_run object, you can poll its status later:

from prefect.deployments import run_deployment
from prefect import get_client
import asyncio

# Trigger without waiting
flow_run = run_deployment(
    name="custom-movements-flow/custom-movements",
    parameters={"position_inches": 35.5},
    timeout=0
)

# Later... poll the status
async def check_status(flow_run_id):
    async with get_client() as client:
        flow_run = await client.read_flow_run(flow_run_id)
        print(f"Status: {flow_run.state.type}")
        print(f"Result: {flow_run.state.result if flow_run.state_type == 'COMPLETED' else 'Still running'}")
        return flow_run

# Check status anytime
flow_run_status = asyncio.run(check_status(flow_run.id))
Key Parameters:

timeout=0: Returns immediately without waiting for completion
timeout=300: Waits up to 5 minutes for completion
await run_deployment(): Non-blocking async version
This gives you exactly what you need—send the command (preheat to 350°), get back the flow run ID immediately, then check on it whenever you want!